package watermaker;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Create by zhh on 2022/9/14.
 * EventTime结合WM的使用
 * 输入数据：时间字段，单词，次数
 */
public class EventTimeWMAPP {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        test01(env);

        env.execute("WMApp");
    }
    public static void test01(StreamExecutionEnvironment env){
        //延迟数据的标签
        OutputTag<Tuple2<String,Integer>> outputTag = new OutputTag<Tuple2<String,Integer>>("late-data"){};

        SingleOutputStreamOperator<String> lines = env.socketTextStream("192.168.32.222", 9527)
                .assignTimestampsAndWatermarks(
                        new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {//最大允许延迟的时间
                            @Override
                            //设置EventTime的时间字段
                            public long extractTimestamp(String element) {
                                return Long.parseLong(element.split(",")[0]);//获取输入数据的第一个字段，即时间字段
                            }
                        }
                );
        KeyedStream<Tuple2<String, Integer>, String> keyedStream = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                String[] splits = s.split(",");
                return Tuple2.of(splits[1].trim(), Integer.parseInt(splits[2].trim()));
            }
        }).keyBy(x -> x.f0);
        //[0000,5000),[5000,10000)
        SingleOutputStreamOperator<String> window = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5)))//EventTime的滚动窗口，5秒滚动
                .sideOutputLateData(outputTag)//调用延迟数据
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                        System.out.println("------reduce执行-----\n (" + value1.f0 + "," + (value1.f1 + value2.f1) + ")");
                        return Tuple2.of(value1.f0, value1.f1 + value2.f1);
                    }
                }, new ProcessWindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {
                    FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");//设置时间打印的格式

                    @Override
                    public void process(String s, Context context, Iterable<Tuple2<String, Integer>> elements, Collector<String> out) throws Exception {
                        for (Tuple2<String, Integer> element : elements) {
                            //wm的时间
                            System.out.println("WMTime:[" + format.format(context.currentWatermark()) + "]");
                            //打印窗口开始和结束的时间以及获取的值
                            out.collect("[" + this.format.format(context.window().getStart()) + "==>" + this.format.format(context.window().getEnd()) + "],(" + element.f0 + "," + element.f1 + ")");
                        }
                    }
                });
        window.print();
        DataStream<Tuple2<String, Integer>> sideOutput = window.getSideOutput(outputTag);//获取延迟的数据
        sideOutput.printToErr();//打印延迟的数据
    }
}
